#include <poll.h>
#include <sys/types.h>
#include <sys/socket.h>

#include <string.h>
#include <assert.h>
#include <mutex>
#include <algorithm>

#include "fd_reactor.h"
#include "hook_helper.h"
#include "../co_api.h"
#include "../co_utils.h"
#include "../common/net_utils.h"
#include "../common/common_utils.h"

using namespace std;

const int ERR_EVENTS = EPOLLHUP | EPOLLRDHUP | EPOLLERR;

#define NOTIFY_READ  0
#define NOTIFY_WRITE 1

inline long ptr_to_long(void* ptr)
{
    return reinterpret_cast<long>(ptr);
}

CoFdReactor::CoFdReactor()
{
    init();
}

CoFdReactor::~CoFdReactor()
{
    LOG_DEBUG("~CoFdReactor");
	stop();
}

bool CoFdReactor::init()
{
    int i;
    for (i = 0; i < g_manager.get_env_param().epoll_worker_count; i++) {
		auto ptr = shared_ptr<CoReactorInfo>(new CoReactorInfo);
        ptr->epoll_id = epoll_create(10000);
        if (ptr->epoll_id == -1) {
            LOG_ERROR("epoll_create fail, index:%d, errno:%d", i, errno);
            break ; 
        }
        if (pipe(ptr->notify_fds) != 0) {
            LOG_ERROR("pipe fail, index:%d, errno:%d", i, errno);
            break ;
        }
        for (int j = 0; j < MAX_SLOT_COUNT; j++) {
            auto slot_ptr = shared_ptr<CoReactorSlotInfo>(new CoReactorSlotInfo);
            ptr->slots.push_back(slot_ptr);
        }
        _reactor_infos.push_back(ptr);
    }

    if ((int)_reactor_infos.size() != g_manager.get_env_param().epoll_worker_count) {
        std::for_each(
            _reactor_infos.begin(),        
            _reactor_infos.end(),        
            [](const shared_ptr<CoReactorInfo>& info) {
                close(info->epoll_id); 
                close(info->notify_fds[0]);
                close(info->notify_fds[1]);
            }
        ); 
        return false;
    }
   
    for (int i = 0; i < (int)_reactor_infos.size(); i++) {
		auto t = thread([this, i] {run(i);});
		pthread_setname_np(
			t.native_handle(), 
			CommonUtils::format_string("reactor_%d", i).c_str()
		);
		_threads.emplace_back(move(t));
    }

    _counter.set_field_buckets(CoFdReactorCounter::EPOLL_WAIT_COUNTER, g_manager.get_env_param().epoll_worker_count);
    _counter.set_field_buckets(CoFdReactorCounter::EPOLL_WAIT_COST_COUNTER, g_manager.get_env_param().epoll_worker_count);
    _counter.set_field_buckets(CoFdReactorCounter::EPOLL_HANDLE_FD_COUNTER, g_manager.get_env_param().epoll_worker_count);

    _counter.set_field_buckets(CoFdReactorCounter::EPOLL_MOD_COUNTER, g_manager.get_env_param().epoll_worker_count);
    _counter.set_field_buckets(CoFdReactorCounter::EPOLL_MOD_COST_COUNTER, g_manager.get_env_param().epoll_worker_count);

    _counter.set_field_buckets(CoFdReactorCounter::EPOLL_ADD_COUNTER, g_manager.get_env_param().epoll_worker_count);
    _counter.set_field_buckets(CoFdReactorCounter::EPOLL_ADD_COST_COUNTER, g_manager.get_env_param().epoll_worker_count);

    _counter.set_field_buckets(CoFdReactorCounter::EPOLL_DEL_COUNTER, g_manager.get_env_param().epoll_worker_count);
    _counter.set_field_buckets(CoFdReactorCounter::EPOLL_DEL_COST_COUNTER, g_manager.get_env_param().epoll_worker_count);

    _counter.init(g_manager.is_record_counter());
   // _counter.init(false);

    return true; 
}

void CoFdReactor::stop()
{
    LOG_DEBUG("fd_reactor stop start");
    for (auto& item : _reactor_infos) {
        close(item->notify_fds[NOTIFY_WRITE]);
        LOG_DEBUG("fd_reactor close notify_id:%d", item->notify_fds[NOTIFY_WRITE]);
    }
    for (auto& item : _threads) {
        item.join();
		LOG_DEBUG("reactor_thread is stop");
    }
	for (auto& item : _reactor_infos) {
		close(item->epoll_id);
		LOG_DEBUG("fd_reactor close epoll_id:%d", item->epoll_id);
	}
	LOG_DEBUG("fd_reactor stop finish");
}

bool CoFdReactor::is_exist(FdContextPtr ptr)
{
    int fd = ptr->get_fd();
    if (fd < 0) {
        LOG_ERROR("fd invalid, fd:%d", fd);
        return false;
    }

    int index = fd % _reactor_infos.size();
    int slot  = fd % MAX_SLOT_COUNT;

    auto reactor_info = _reactor_infos[index];
    auto slot_ptr     = reactor_info->slots[slot];

    auto iter = slot_ptr->fd_infos.find(fd);

    return iter != slot_ptr->fd_infos.end() ? true : false;
}

bool CoFdReactor::set(FdContextPtr ptr, short events, short* revents, const CoSuspendEntry& entry)
{
    int fd = ptr->get_fd();
    if (fd < 0) {
        LOG_ERROR("fd invalid, fd:%d", fd);
        return false;
    }

    if (!(events & (EPOLLIN | EPOLLOUT))) {
        LOG_ERROR("events invalid, fd:%d, events:%d", fd, events);
        return false;
    }

	int index = fd % _reactor_infos.size();
    int slot  = fd % MAX_SLOT_COUNT;

    auto reactor_info = _reactor_infos[index];
    auto slot_ptr = reactor_info->slots[slot];

    int  set_op = 0;
    long set_op_cost = 0;
    {
        LOCK_GUARD(mutex, slot_ptr->mu);

        auto iter = slot_ptr->fd_infos.find(fd);
        set_op = 
            iter != slot_ptr->fd_infos.end() ? 
            EPOLL_CTL_MOD : 
            EPOLL_CTL_ADD;

        struct epoll_event ev;
        ev.data.fd = fd;
        ev.events  = ptr->get_events() | events | EPOLLONESHOT;

        auto beg_us = CommonUtils::now_us();
        int ret = epoll_ctl(reactor_info->epoll_id, set_op, fd, &ev);
        auto end_us = CommonUtils::now_us();
        if (ret) {
            LOG_ERROR(
                "epoll_ctl fail, epoll_id:%d, op:%s, fd:%d, ptr:%p, errno:%d", 
                reactor_info->epoll_id,
                set_op == EPOLL_CTL_MOD ? "mod" : "add",
                fd, 
                ptr.get(),
                errno
            );
            return false;
        }
        set_op_cost = end_us - beg_us;

        auto& fd_info = slot_ptr->fd_infos[fd];

        fd_info.ptr = ptr;
        if (events & EPOLLIN) {
            fd_info.r_cid_infos[entry.cid] = {revents, entry};
        }
        if (events & EPOLLOUT) {
            fd_info.w_cid_infos[entry.cid] = {revents, entry};
        }
    }

    if (set_op == EPOLL_CTL_MOD) {
        _counter.incr(CoFdReactorCounter::EPOLL_MOD_COUNTER, index);
        _counter.add(CoFdReactorCounter::EPOLL_MOD_COST_COUNTER, set_op_cost, index);
    } else {
        _counter.incr(CoFdReactorCounter::EPOLL_ADD_COUNTER, index);
        _counter.add(CoFdReactorCounter::EPOLL_ADD_COST_COUNTER, set_op_cost, index);
    }

    return true;
}

bool CoFdReactor::detach(FdContextPtr ptr, std::function<void (FdContextPtr&, bool)> func)
{
    int fd = ptr->get_fd();
    if (fd < 0) {
        LOG_ERROR("fd invalid, fd:%d", fd);
        return false;
    }

    int index = fd % _reactor_infos.size();
    int slot  = fd % MAX_SLOT_COUNT;

    auto reactor_info = _reactor_infos[index];
    auto slot_ptr = reactor_info->slots[slot];

    LOCK_GUARD(mutex, slot_ptr->mu);

    auto iter = slot_ptr->fd_infos.find(fd);
    if (iter == slot_ptr->fd_infos.end()) {
        LOG_WARN("slot not found fd, fd:%d, reactor:%d, slot:%d", fd, index, slot);
        return false;
    }

    bool is_release = false;

    auto& fd_info = iter->second;
    if (!fd_info.r_cid_infos.size() && !fd_info.w_cid_infos.size())
    {
        int ret = epoll_ctl(reactor_info->epoll_id, EPOLL_CTL_DEL, fd, NULL);
        if (ret) {
            LOG_WARN(
                "epoll_ctl fail, worker:%d, epoll_id:%d, op:del, fd:%d, ptr:%p, errno:%d", 
                index, 
                reactor_info->epoll_id, 
                fd, 
                ptr.get(),
                errno
            );
			assert(false);
        }
        slot_ptr->fd_infos.erase(fd);
        is_release = true;
    }
    if (func) {
        func(ptr, is_release);
    }
    return true;
}

void CoFdReactor::run(int index)
{
    const int DEF_EVENT_SIZE = 10000;

    struct epoll_event events[DEF_EVENT_SIZE];

    int epoll_id  = _reactor_infos[index]->epoll_id;
    int notify_fd = _reactor_infos[index]->notify_fds[NOTIFY_READ]; 

    LOG_DEBUG(
        "fd_reactor worker running, worker:%d, epoll_id:%d, notify_fd:%d", 
        index, 
        epoll_id, 
        notify_fd
    );

    {
        struct epoll_event ev;
        ev.data.fd = notify_fd;
        ev.events  = EPOLLIN;

        int ret = epoll_ctl(epoll_id, EPOLL_CTL_ADD, notify_fd, &ev);
        if (ret) {
            LOG_ERROR(
                "epoll_ctl fail, worker:%d, epoll_id:%d, notify_fd:%d, errno:%d", 
                index,
                epoll_id,
                notify_fd,
                errno
            );
        }
        assert(!ret);
    }

    while (1) {

        auto beg_us = CommonUtils::now_us();
        int count = CommonUtils::call_without_intr(
            epoll_wait,
            epoll_id,
            events,
            DEF_EVENT_SIZE,
            -1
        );
        auto end_us = CommonUtils::now_us();

        if (count == -1) {
            LOG_ERROR(
                "epoll_wait fail, worker:%d, epoll_id:%d, errno:%d", 
                index,
                epoll_id,
                errno
            );
        }
        assert(count >= 0);

        _counter.incr(CoFdReactorCounter::EPOLL_WAIT_COUNTER, index);
        _counter.add(CoFdReactorCounter::EPOLL_WAIT_COST_COUNTER, end_us - beg_us, index);
        _counter.add(CoFdReactorCounter::EPOLL_HANDLE_FD_COUNTER, count, index);

        for (int i = 0; i < count; i++) {

            int fd = events[i].data.fd;
            int ev = events[i].events;

            if (fd == notify_fd) {
                LOG_INFO("worker exit, worker:%d", index);
            } else {
                handle_event(index, fd, ev);
            }
        }
    }
}

void CoFdReactor::handle_event(int index, int fd, int events)
{
    auto reactor_info = _reactor_infos[index];

    int  epoll_id = reactor_info->epoll_id;
    auto slot_ptr = reactor_info->slots[fd % MAX_SLOT_COUNT];

    CoReactorFdInfo* fd_info_ptr = NULL;
    std::map<int, CoFdCidInfo> r_cid_infos;
    std::map<int, CoFdCidInfo> w_cid_infos;
    {
        LOCK_GUARD(mutex, slot_ptr->mu);

        auto iter = slot_ptr->fd_infos.find(fd);
        if (iter == slot_ptr->fd_infos.end()) {
            LOG_ERROR("not found fd_info_ptr in slots, worker:%d, fd:%d", index, fd);
            assert(false);
            return ;
        }

        fd_info_ptr = &(iter->second);

        if (events & EPOLLIN) {
            std::swap(fd_info_ptr->r_cid_infos, r_cid_infos);
        }
        if (events & EPOLLOUT) {
            std::swap(fd_info_ptr->w_cid_infos, w_cid_infos);
        }
    }

    int  set_op = 0;
    long cost_us = 0;
    {
        LOCK_GUARD(mutex, slot_ptr->mu);

        if (fd_info_ptr->ptr.use_count() > 1) {
            int new_events = 0;
            if (fd_info_ptr->r_cid_infos.size()) {
                new_events |= EPOLLIN;
            }
            if (fd_info_ptr->w_cid_infos.size()) {
                new_events |= EPOLLOUT;
            }
            if (new_events) {

                struct epoll_event ev;
                ev.data.fd = fd;
                ev.events  = new_events | EPOLLONESHOT;

                auto beg_us = CommonUtils::now_us();
                int ret = epoll_ctl(epoll_id, EPOLL_CTL_MOD, fd, &ev);
                auto end_us = CommonUtils::now_us();
                if (!ret) {
                    set_op  = EPOLL_CTL_MOD;
                    cost_us = end_us - beg_us;
                } else {
                    LOG_ERROR(
                        "epoll_ctl fail, worker:%d, epoll_id:%d, op:mod, fd:%d, events:%d, errno:%d", 
                        index, 
                        epoll_id,
                        fd,
                        ev.events,
                        errno
                    );
                }
                assert(ret == 0);
            }
        } else {
            auto beg_us = CommonUtils::now_us();
            int ret = epoll_ctl(epoll_id, EPOLL_CTL_DEL, fd, NULL);
            auto end_us = CommonUtils::now_us();
            if (!ret) {
                set_op  = EPOLL_CTL_DEL;
                cost_us = end_us - beg_us;
            } else {
                LOG_ERROR(
                    "epoll_ctl fail, worker:%d, epoll_id:%d, op:del, fd:%d, errno:%d", 
                    index,
                    epoll_id,
                    fd,
                    errno
                );
            }
            assert(ret == 0);
            slot_ptr->fd_infos.erase(fd);
        }

        if (set_op == EPOLL_CTL_MOD) {
            _counter.incr(CoFdReactorCounter::EPOLL_MOD_COUNTER, index);
            _counter.add(CoFdReactorCounter::EPOLL_MOD_COST_COUNTER, cost_us, index);
        } else {
            _counter.incr(CoFdReactorCounter::EPOLL_DEL_COUNTER, index);
            _counter.add(CoFdReactorCounter::EPOLL_DEL_COST_COUNTER, cost_us, index);
        }

        auto active_coroutine = [this, index, fd](std::map<int, CoFdCidInfo>& cid_infos, short events) {
            for (auto& item : cid_infos) {
                int   cid  = item.first;
                auto& info = item.second;
                *info.revents = events;
                g_manager.resume_co_by_suspend_entry(info.entry);
                LOG_DEBUG(
                    "reactor event wake coroutine, worker:%d, fd:%d, cid:%d, revents:%d", 
                    index,
                    fd,
                    cid,
                    events
                );
            }
        };
        active_coroutine(r_cid_infos, events & EPOLLIN);
        active_coroutine(w_cid_infos, events & EPOLLOUT);
    }
}

CoFdReactorCounter CoFdReactor::get_counter_statics()
{
	return _counter.get();
}

